package com.chinamobile.iot.command;

import com.alibaba.fastjson.JSON;
import com.chinamobile.iot.config.KafkaTopicConfiguration;
import com.chinamobile.iot.kafka.KafkaCommandProducer;
import com.chinamobile.iot.mqtt.spi.processor.CommandState;
import com.chinamobile.iot.mqtt.spi.processor.DataProcessor;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.RemovalCause;
import com.github.benmanes.caffeine.cache.RemovalListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * Created by szl on 2017/5/27.
 */
@Component
public class CommandAckCacheTimeoutListener implements RemovalListener<CommandKey, String> {

    private static Logger logger = LoggerFactory.getLogger(CommandAckCacheTimeoutListener.class);
    @Autowired
    private KafkaTopicConfiguration kafkaTopic;
    @Autowired
    private KafkaCommandProducer kafkaCommandProducer;
    @Autowired
    private DataProcessor dataProcessor;
    private Cache<CommandKey, String> cacheCommand;

    @Override
    public void onRemoval(CommandKey key, String value, RemovalCause cause) {
        if (cause == RemovalCause.EXPIRED) {
            cacheCommand.invalidate(key);
            String clientId = key.getClientId();
            String mqttId = key.getMqttId();
            String commandId = value;

            Object cmd = dataProcessor.createCommandEvent(clientId, mqttId, commandId, CommandState.ACKED);
            if (cmd != null) {
                kafkaCommandProducer.sendCommand(mqttId, kafkaTopic.cmd(), cmd);
                logger.info("send cmd ack to kafka : {}", JSON.toJSONString(cmd));
            }
        }
    }

    public void setCacheCommand(Cache<CommandKey, String> cacheCommand) {
        this.cacheCommand = cacheCommand;
    }
}
